package com.nami.gate.middleware.mq.impl;


import com.nami.gate.config.rocketmq.ConsumerConfig;
import com.nami.gate.config.rocketmq.RocketMqConfig;
import com.nami.gate.external.basic.constants.DownTopicConst;
import com.nami.gate.middleware.mq.factory.ConsumerConfigWrap;
import com.nami.gate.middleware.mq.factory.RocketMqConsumerFactory;
import com.nami.gate.middleware.mq.MqConsumer;
import com.nami.gate.middleware.mq.listener.*;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;


public class RocketMqConsumerImpl implements MqConsumer {

    private static final Logger log = LoggerFactory.getLogger(RocketMqConsumerImpl.class);

    private RocketMqConfig config;


    public RocketMqConsumerImpl(RocketMqConfig config) {
        this.config = config;
    }

    private List<DefaultMQPushConsumer> consumers;

    @Override
    public void start() throws Exception {
        // 初始化mq receiver
        // 1.初始化 消费者组，topic, tags, 回调invoke
        // 2.组成list
        // 3.循环创建多个消费者订阅

        ConsumerConfig consumerConfig = config.getConsumerConfig();

        List<ConsumerConfigWrap> wrappers = new ArrayList<>();
        ConsumerConfigWrap login = new ConsumerConfigWrap();
        login.setTopic(DownTopicConst.LOGIN_TOPIC);
        login.setTags(config.getTags());
        login.setConsumerGroup(consumerConfig.getLoginConsumerGroup());
        login.setMessageListener(new DownMsgLoginConsumer());
        login.setNamesrvAddr(config.getNamesrvAddr());
        login.setMessageModel(MessageModel.CLUSTERING);
        wrappers.add(login);

        ConsumerConfigWrap logout = new ConsumerConfigWrap();
        logout.setNamesrvAddr(config.getNamesrvAddr());
        logout.setTopic(DownTopicConst.LOGOUT_TOPIC);
        logout.setTags(config.getTags());
        logout.setConsumerGroup(consumerConfig.getLogoutConsumerGroup());
        logout.setMessageModel(MessageModel.CLUSTERING);
        logout.setMessageListener(new DownMsgLogOutConsumer());
        wrappers.add(logout);

        ConsumerConfigWrap closeChannel = new ConsumerConfigWrap();
        closeChannel.setNamesrvAddr(config.getNamesrvAddr());
        closeChannel.setTopic(DownTopicConst.CLOSE_CHANNEL_TOPIC);
        closeChannel.setTags(config.getTags());
        closeChannel.setConsumerGroup(consumerConfig.getCloseConsumerGroup());
        closeChannel.setMessageModel(MessageModel.CLUSTERING);
        closeChannel.setMessageListener(new DownMsgCloseChannelConsumer());
        wrappers.add(closeChannel);

        ConsumerConfigWrap flushOne = new ConsumerConfigWrap();
        flushOne.setNamesrvAddr(config.getNamesrvAddr());
        flushOne.setTopic(DownTopicConst.FLUSH_ONE_TOPIC);
        flushOne.setTags(config.getTags());
        flushOne.setConsumerGroup(consumerConfig.getFlushOneConsumerGroup());
        flushOne.setMessageModel(MessageModel.CLUSTERING);
        flushOne.setMessageListener(new DownMsgFlushOneConsumer());
        wrappers.add(flushOne);

        ConsumerConfigWrap flushMore = new ConsumerConfigWrap();
        flushMore.setNamesrvAddr(config.getNamesrvAddr());
        flushMore.setTopic(DownTopicConst.FLUSH_MORE_TOPIC);
        flushMore.setTags(config.getTags());
        flushMore.setMessageModel(MessageModel.CLUSTERING);
        flushMore.setConsumerGroup(consumerConfig.getFlushMoreConsumerGroup());
        flushMore.setMessageListener(new DownMsgFlushBatchConsumer());
        wrappers.add(flushMore);

        consumers = new ArrayList<>(wrappers.size());
        for (ConsumerConfigWrap wrapper : wrappers) {
            DefaultMQPushConsumer consumer = RocketMqConsumerFactory.startConsumer(wrapper);
            log.info("mq consumer start success. topic={}|{}, consumerGroup:{}, mode:{}", wrapper.getTopic(), wrapper.getTags(), wrapper.getConsumerGroup(), wrapper.getMessageModel());
            consumers.add(consumer);
        }
    }

    @Override
    public void shutdown() {
        for (DefaultMQPushConsumer consumer : consumers) {
            if (consumer != null) {
                consumer.shutdown();
            }
        }
    }
}
